package com.yanggu.spark.core.accumulator

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable

//自定义累加器
//需要继承自AccumulatorV2, 重写6个抽象方法, 同时需要给定输入、输出的泛型
//核心方法为add和merge方法
class WordCountAccumulator extends AccumulatorV2[String, mutable.Map[String, Int]] {

  var map: mutable.Map[String, Int] = mutable.Map()

  override def isZero = map.isEmpty

  override def copy() = new WordCountAccumulator

  override def reset(): Unit = map.clear()

  override def add(word: String): Unit = {
    map(word) = map.getOrElse(word, 0) + 1
  }

  //这里是将两个map的数据进行合并
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
    //将其他task的累加器的数据合并到当前的累加器中, 也就是map中
    // 两个Map的合并
    //这里需要注意的是不能使用foreach。因为有可能某个遍历的map为空, 导致循环根本没有执行
    //这里需要使用foldLeft折叠。
    map = map.foldLeft(other.value) {
      case (map2, (word, count)) =>
        map2 + (word -> (count + map2.getOrElse(word, 0)))
    }
  }

  override def value = map
}
